package main

import (
	"fmt"
	"os"
	"os/signal"
	"reflect"
	"strconv"
	"syscall"
	"time"
)

func main() {
	//testMultiChan()
	//testMultiChanWithReflect()
	//testDataTransfer()
	//testSignalNotice()
	//testMutexLock()
	testOrDone()

}

// testMultiChan
//
//	@Description: 这里只有两个chan，如果有n个如何处理,将testMultiChanWithReflect
func testMultiChan() {
	const (
		sendDelay   = 1 * time.Second
		timeoutTime = 3 * time.Second
	)

	var (
		ch1 = make(chan string, 1)
		ch2 = make(chan string, 1)
	)

	time.AfterFunc(sendDelay, func() { ch1 <- "channel-1:发送的数据" })
	time.AfterFunc(sendDelay, func() { ch2 <- "channel-2:发送的数据" })

	timer := time.NewTimer(timeoutTime)
	defer timer.Stop() // 确保定时器资源回收

	select {
	case v := <-ch1:
		fmt.Println(v)
	case v := <-ch2:
		fmt.Println(v)
	case <-timer.C:
		fmt.Println("time out")
	}
}

func testMultiChanWithReflect() {
	var ch1 = make(chan string, 10)
	var ch2 = make(chan string, 10)

	// 创建SelectCase
	var cases = createCases(ch1, ch2)

	// 执行10次select
	for i := 0; i < 10; i++ {
		//chosen-被触发的 case 在传入的 cases 切片中的索引
		//recv-接收操作获取到的值（仅当 case 是 SelectRecv 时有效）
		//ok-接收操作是否成功（类似 val, ok := <-ch 中的 ok）
		chosen, recv, ok := reflect.Select(cases)
		if recv.IsValid() { // 接收操作
			fmt.Println("接收的数据:", cases[chosen].Dir, recv, ok)
		} else { // 发送操作
			fmt.Println("发送的数据:", cases[chosen].Dir, ok)
		}
	}
}

//	type SelectCase struct {
//		Dir  SelectDir   // 操作方向：Send/Recv/Default
//		Chan Value       // 操作的 channel（反射类型） channel 的反射值
//		Send Value       // 发送的值（仅当 Dir=Send 时有效）
//	}
func createCases(chs ...chan string) []reflect.SelectCase {
	// 创建SelectCase切片
	var cases []reflect.SelectCase

	// 创建接收 case
	for _, recvChan := range chs {
		cases = append(cases, reflect.SelectCase{
			Dir:  reflect.SelectRecv,        //接收枚举
			Chan: reflect.ValueOf(recvChan), //chan的反射Value
		})
	}

	// 创建发送 case
	for i, sendChan := range chs {
		v := reflect.ValueOf("channel-" + strconv.Itoa(i) + ":发送的数据")
		cases = append(cases, reflect.SelectCase{
			Dir:  reflect.SelectSend,        //发送枚举
			Chan: reflect.ValueOf(sendChan), //chan的反射Value
			Send: v,                         //发送的数据
		})
	}

	return cases
}

// testDataTransfer
//
//	@Description: 数据传递
func testDataTransfer() {

	chs := []chan Token{make(chan Token), make(chan Token), make(chan Token), make(chan Token)}

	// 创建4个worker
	for id := 0; id < 4; id++ {
		//id-序号
		//ch-当前worker的接收通道
		//nextCh-下一个worker的接收通道
		go newWorker(id, chs[id], chs[(id+1)%4])
	}

	//首先把令牌交给第一个worker
	chs[0] <- struct{}{}

	select {}
}

type Token struct{}

func newWorker(id int, ch chan Token, nextCh chan Token) {
	for {
		token := <-ch         // 取得令牌
		fmt.Println((id + 1)) // id从1开始
		time.Sleep(time.Second)
		nextCh <- token
	}
}

// testSignalNotice
//
//	@Description: 测试信号通知
func testSignalNotice() {
	go func() {
		for {
			time.Sleep(time.Second)
			fmt.Println("do something ...")
			// 发送信号
			//signal.Notify(os.Stdin, os.Interrupt)
		}
	}()

	// 处理CTRL+C等中断信号
	termChan := make(chan os.Signal)
	signal.Notify(termChan, syscall.SIGINT, syscall.SIGTERM)
	<-termChan

	func() {
		fmt.Println("做一些清理工作")
	}()
	fmt.Println("优雅退出")
}

// testMutexLock
//
//	@Description: 使用chan实现mutex
func testMutexLock() {
	m := NewMutex()
	ok := m.TryLock()
	fmt.Printf("locked v %v\n", ok)
	ok = m.TryLock()
	fmt.Printf("locked %v\n", ok)
	println(m.IsLocked())
	m.Unlock()
	m.Unlock()

}

type Mutex struct {
	ch chan int
}

// NewMutex 使用锁需要初始化
func NewMutex() *Mutex {
	mu := &Mutex{ch: make(chan int, 1)}
	mu.ch <- 1
	return mu
}

func (m *Mutex) Lock() {
	<-m.ch
}

func (m *Mutex) Unlock() {
	select {
	case m.ch <- 1:
		println("unlock")
	default:
		panic("unlock of unlocked mutex")
	}
}

func (m *Mutex) TryLock() bool {
	select {
	case <-m.ch:
		return true
	default:
		return false
	}
}
func (m *Mutex) TryLockWithTimeout(timeout time.Duration) bool {
	//定时器
	timer := time.NewTimer(timeout)
	defer timer.Stop()

	select {
	case <-m.ch:
		return true
	case <-timer.C:
		return false
	}
}

func (m *Mutex) IsLocked() bool {
	return len(m.ch) == 0
}

// testOrDone
//
//	@Description: 这是有一个任务的情况，如果有多个任务，只要有任意一个任务执行完，我们就想获得这个信号，这就是 Or-Done 模式。
func testOrDone() {
	start := time.Now()
	chan01 := sig(10 * time.Second)
	chan02 := sig(15 * time.Second)
	chan03 := sig(20 * time.Second)
	chan04 := sig(5 * time.Second)
	chan05 := sig(25 * time.Second)
	chan06 := sig(30 * time.Second)

	<-orDoneReflect(chan01, chan02, chan03, chan04, chan05, chan06)
	fmt.Printf("done after %v", time.Since(start))
}

// sig
//
//	@Description: 注册信号，通道将在 after 时间后关闭,关闭后可以零值和false
//	@param after
//	@return <-happensBeforeChan 只读channel
func sig(after time.Duration) <-chan interface{} {
	c := make(chan interface{})
	go func() {
		defer close(c)
		time.Sleep(after)
	}()
	return c
}

// orDone
//
//	@Description: or-Done:使用递归
//	@param channels
//	@return <-happensBeforeChan
func orDoneRecursion(channels ...<-chan interface{}) <-chan interface{} {
	if nil == channels || len(channels) == 0 {
		panic("can not be zero channels")
	}
	//只有一个channel，直接返回，不需要再单独开协程进行处理
	if len(channels) == 1 {
		return channels[0]
	}
	//>=2个channel，使用协程+select处理
	orDoneChan := make(chan interface{})
	go func() {
		defer close(orDoneChan)
		switch len(channels) {
		case 2: // 2个也是一种特殊情况，任意一个通道关闭（读取到零值和false）即返回
			select {
			case <-channels[0]:
			case <-channels[1]:
			}
		default: //超过两个，二分法递归处理
			m := len(channels) / 2
			select {
			case <-orDoneRecursion(channels[:m]...):
			case <-orDoneRecursion(channels[m:]...):
			}
		}
	}()
	return orDoneChan
}

// orDoneReflect
//
//	@Description: or-Done:使用反射
//	@param channels
//	@return <-happensBeforeChan
func orDoneReflect(channels ...<-chan interface{}) <-chan interface{} {
	if nil == channels || len(channels) == 0 {
		panic("can not be zero channels")
	}
	//只有一个channel，直接返回，不需要再单独开协程进行处理
	if len(channels) == 1 {
		return channels[0]
	}

	orDoneChan := make(chan interface{})
	go func() {
		defer close(orDoneChan)
		// 利用反射构建SelectCase
		var cases []reflect.SelectCase
		for _, c := range channels {
			cases = append(cases, reflect.SelectCase{
				Dir:  reflect.SelectRecv,
				Chan: reflect.ValueOf(c),
			})
		}
		// 随机选择一个可用的case
		reflect.Select(cases)
	}()

	return orDoneChan
}

// fanInReflect
//
//	@Description: 扇入模式-反射实现
//	在软件工程中，模块的扇入是指有多少个上级模块调用它。而对于我们这里的 Channel 扇入模式来说，就是指有多个源 Channel 输入、一个目的 Channel 输出的情况。
//	扇入比就是源 Channel 数量比 1。每个源 Channel 的元素都会发送给目标 Channel，相当于目标 Channel 的 receiver 只需要监听目标 Channel，就可以接收所有发送给源 Channel 的数据。
//	@param chans
//	@return <-happensBeforeChan
func fanInReflect(chans ...<-chan interface{}) <-chan interface{} {
	out := make(chan interface{})
	go func() {
		defer close(out)
		// 构造SelectCase slice
		var cases []reflect.SelectCase
		for _, c := range chans {
			cases = append(cases, reflect.SelectCase{
				Dir:  reflect.SelectRecv,
				Chan: reflect.ValueOf(c),
			})
		}

		// 循环，从cases中选择一个可用的
		for len(cases) > 0 {
			i, v, ok := reflect.Select(cases)
			if !ok { // 此channel已经close
				cases = append(cases[:i], cases[i+1:]...)
				continue
			}
			out <- v.Interface()
		}
	}()
	return out
}

// fanInRec
//
//	@Description: 扇入模式-递归实现
//	@param chans
//	@return <-happensBeforeChan
func fanInRec(chans ...<-chan interface{}) <-chan interface{} {
	switch len(chans) {
	case 0:
		c := make(chan interface{})
		close(c)
		return c
	case 1:
		return chans[0]
	case 2:
		return mergeTwo(chans[0], chans[1])
	default:
		m := len(chans) / 2
		return mergeTwo(
			fanInRec(chans[:m]...),
			fanInRec(chans[m:]...))
	}
}

func mergeTwo(a, b <-chan interface{}) <-chan interface{} {
	c := make(chan interface{})
	go func() {
		defer close(c)
		for a != nil || b != nil { //只要还有可读的chan
			select {
			case v, ok := <-a:
				if !ok { // a 已关闭，设置为nil
					a = nil
					continue
				}
				c <- v
			case v, ok := <-b:
				if !ok { // b 已关闭，设置为nil
					b = nil
					continue
				}
				c <- v
			}
		}
	}()
	return c
}

// fanOut
//
//	@Description: 扇出模式只有一个输入源 Channel，有多个目标 Channel，扇出比就是 1 比目标 Channel 数的值，
//
// 经常用在设计模式中的观察者模式中（观察者设计模式定义了对象间的一种一对多的组合关系。
// 这样一来，一个对象的状态发生变化时，所有依赖于它的对象都会得到通知并自动刷新）。
// 在观察者模式中，数据变动后，多个观察者都会收到这个变更信号。
// 下面是一个扇出模式的实现。从源 Channel 取出一个数据后，依次发送给目标 Channel。在发送给目标 Channel 的时候，可以同步发送，也可以异步发送：
//
//	@param ch
//	@param out
//	@param async
func fanOut(ch <-chan interface{}, out []chan interface{}, async bool) {
	go func() {
		defer func() { //退出时关闭所有的输出chan
			for i := 0; i < len(out); i++ {
				close(out[i])
			}
		}()

		for v := range ch { // 从输入chan中读取数据
			v := v
			for i := 0; i < len(out); i++ {
				i := i
				if async { //异步
					go func() {
						out[i] <- v // 放入到输出chan中,异步方式
					}()
				} else {
					out[i] <- v // 放入到输出chan中，同步方式
				}
			}
		}
	}()
}

// asStream
//
//	@Description: Stream
//	这里我来介绍一种把 Channel 当作流式管道使用的方式，也就是把 Channel 看作流（Stream），提供跳过几个元素，或者是只取其中的几个元素等方法。
//	首先，我们提供创建流的方法。这个方法把一个数据 slice 转换成流：
//	@param done
//	@param values
//	@return <-happensBeforeChan
func asStream(done <-chan struct{}, values ...interface{}) <-chan interface{} {
	s := make(chan interface{}) //创建一个unbuffered的channel
	go func() {                 // 启动一个goroutine，往s中塞数据
		defer close(s)             // 退出时关闭chan
		for _, v := range values { // 遍历数组
			select {
			case <-done:
				return
			case s <- v: // 将数组元素塞入到chan中
			}
		}
	}()
	return s
}

// takeN
//
//	@Description: takeFn：筛选流中的数据，只保留满足条件的数据
//	@param done
//	@param valueStream
//	@param num
//	@return <-happensBeforeChan
func takeN(done <-chan struct{}, valueStream <-chan interface{}, num int) <-chan interface{} {
	takeStream := make(chan interface{}) // 创建输出流
	go func() {
		defer close(takeStream)
		for i := 0; i < num; i++ { // 只读取前num个元素
			select {
			case <-done:
				return
			case takeStream <- <-valueStream: //从输入流中读取元素
			}
		}
	}()
	return takeStream
}

// mapChan
//
//	@Description: map-reduce 是一种处理数据的方式
//
// map-reduce 分为两个步骤，第一步是映射（map），处理队列中的数据，
// 第二步是规约（reduce），把列表中的每一个元素按照一定的处理方式处理成结果，放入到结果队列中。
// 就像做汉堡一样，map 就是单独处理每一种食材，reduce 就是从每一份食材中取一部分，做成一个汉堡。
//
//	@param in
//	@param fn
//	@return <-happensBeforeChan
func mapChan(in <-chan interface{}, fn func(interface{}) interface{}) <-chan interface{} {
	out := make(chan interface{}) //创建一个输出chan
	if in == nil {                // 异常检查
		close(out)
		return out
	}

	go func() { // 启动一个goroutine,实现map的主要逻辑
		defer close(out)
		for v := range in { // 从输入chan读取数据，执行业务操作，也就是map操作
			out <- fn(v)
		}
	}()

	return out
}

// reduce
//
//	@Description: reduce 函数的处理逻辑如下
//	@param in
//	@param fn
//	@return interface{}
func reduce(in <-chan interface{}, fn func(r, v interface{}) interface{}) interface{} {
	if in == nil { // 异常检查
		return nil
	}

	out := <-in         // 先读取第一个元素
	for v := range in { // 实现reduce的主要逻辑
		out = fn(out, v)
	}

	return out
}

// 生成一个数据流
func asStream10(done <-chan struct{}) <-chan interface{} {
	s := make(chan interface{})
	values := []int{1, 2, 3, 4, 5}
	go func() {
		defer close(s)
		for _, v := range values { // 从数组生成
			select {
			case <-done:
				return
			case s <- v:
			}
		}
	}()
	return s
}

func testStream() {
	in := asStream10(nil)

	// map操作: 乘以10
	mapFn := func(v interface{}) interface{} {
		return v.(int) * 10
	}

	// reduce操作: 对map的结果进行累加
	reduceFn := func(r, v interface{}) interface{} {
		return r.(int) + v.(int)
	}

	sum := reduce(mapChan(in, mapFn), reduceFn) //返回累加结果
	fmt.Println(sum)
}
